package com.shujia.flink.sink;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Demo1FileSink {
    public static void main(String[] args) throws Exception {
        //1、创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        //2、读取数据
        DataStream<String> studentDS = env.readTextFile("flink/data/students.txt");

        //统计班级的人数
        DataStream<Tuple2<String, Integer>> kvDS = studentDS.map(line -> {
            //取出班级
            String clazz = line.split(",")[4];
            return Tuple2.of(clazz, 1);
        }, Types.TUPLE(Types.STRING, Types.INT));

        //安装班级分组，统计数量
        DataStream<Tuple2<String, Integer>> clazzNumDS = kvDS
                .keyBy(kv -> kv.f0).sum(1);

        //整理格式
        DataStream<String> resultDS = clazzNumDS.map(kv -> kv.f0 + "\t" + kv.f1);

        //将结果保存到文件中
        FileSink<String> fileSink = FileSink.<String>forRowFormat(
                new Path("flink/data/clazz_num"),//保存文件的路径
                new SimpleStringEncoder<>("Utf-8")
        ).build();
        //使用file sink
        resultDS.sinkTo(fileSink);

        env.execute();
    }
}
